package com.ulwx.tool;

import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.log4j.Logger;

import com.ulwx.tool.jms.JmsPoolFactory;

public class JmsUtils {
	private static final Logger log = Logger.getLogger(JmsUtils.class);

	/**
	 * 用于jms的Point-to-Point Domain方式的消息发送
	 * 
	 * @param user
	 * @param password
	 * @param url
	 * @param queueName
	 * @throws Exception
	 */
	public static void send(String poolName, String queueName,
			String sendTextMessage, Map<String,String> properties) throws Exception {

		Connection connection = null;
		Session session = null;
		long start=System.currentTimeMillis();
		try {
			
			connection = JmsPoolFactory.getJmsPool(poolName).createConnection();
			connection.start();
			// The Retailer's session is non-trasacted.
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

			Destination queue = session.createQueue(queueName);
			MessageProducer producer = session.createProducer(queue);
			TextMessage message = session.createTextMessage();
			for(String key: properties.keySet()){
				String val=properties.get(key);
				message.setStringProperty(key, val);
			}
			message.setText(sendTextMessage);
			producer.send(message);

		} catch (JMSException e) {
			log.error("", e);
			throw e;
		} finally {
			if (connection != null) {
				try {
					session.close();
					connection.close();
				} catch (Exception ex) {
					log.error("", ex);
				}
			}
			log.info("sms send:"+(System.currentTimeMillis()-start));
		}
	}

	/**
	 * 用于jms的Point-to-Point Domain方式的消息接收(同步接收)。此方法
	 * 会一直阻塞，除非receiver里的excute方法返回false。
	 * 
	 * @param user
	 * @param password
	 * @param url
	 * @param queueName
	 * @return
	 * @throws Exception
	 */
	public static void receive(String poolName, String queueName,
			Receiver receiver) throws Exception {

		Connection connection = null;
		Session session = null;
		Destination destination = null;

		try {

			connection = JmsPoolFactory.getJmsPool(poolName).createConnection();
			connection.start();
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			destination = session.createQueue(queueName);
			MessageConsumer consumer = session.createConsumer(destination);

			// consumer.setMessageListener(receiver);

			while (true) {
				TextMessage message = (TextMessage) consumer.receive();
				String msg = message.getText();
				Map<String,String> properties=new HashMap<String,String>();
				Enumeration  names=message.getPropertyNames();
				while(names.hasMoreElements()){
					String name=(String)names.nextElement();
					properties.put(name, message.getStringProperty(name));
				}
				boolean result = receiver.excute(msg,properties);
				if (!result) {
					break;
				}
			}
		} catch (Exception e) {
			log.error("", e);
			throw e;
		} finally {

			if (connection != null) {
				try {
					session.close();
					connection.close();
				} catch (Exception ex) {
					log.error("", ex);
				}
			}
		}

	}

	/**
	 * 用于jms的Point-to-Point Domain方式的消息接收(异步接收)。此方法
	 * 会一直阻塞，除非receiver里的excute方法返回false
	 * 
	 * @param user
	 * @param password
	 * @param url
	 * @param queueName
	 * @return
	 * @throws Exception
	 */
	public static void receiveAsyn(String poolName, String queueName,
			final Receiver Receiver) throws Exception {

		Connection connection = null;
		Session session = null;
		Destination destination = null;
		final Object synLock = new Object();
		try {

			// connection = pooledConnectionFactory.createConnection();
			connection = JmsPoolFactory.getJmsPool(poolName).createConnection();
			connection.start();
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			destination = session.createQueue(queueName);
			MessageConsumer consumer = session.createConsumer(destination);

			consumer.setMessageListener(new MessageListener() {

				@Override
				public void onMessage(Message msg) {
					// TODO Auto-generated method stub
					if (msg instanceof TextMessage) {
						TextMessage textMessage = (TextMessage) msg;
						try {
							String txtMsg = textMessage.getText();
							Map<String,String> properties=new HashMap<String,String>();
							Enumeration  names=textMessage.getPropertyNames();
							while(names.hasMoreElements()){
								String name=(String)names.nextElement();
								properties.put(name, textMessage.getStringProperty(name));
							}
							boolean result = Receiver.excute(txtMsg,properties);
							if (!result) {
								synchronized (synLock) {
									synLock.notify();
								}
							}
						} catch (JMSException e) {
							log.error("", e);
						}

					} else {
						log.error("消息不是文本消息！");
					}
				}

			});

			synchronized(synLock) {
				synLock.wait();
			}
		} catch (Exception e) {
			log.error("", e);
			throw e;
		} finally {

			if (connection != null) {
				try {
					session.close();
					connection.close();
				} catch (Exception ex) {
					log.error("", ex);
				}
			}
		}

	}

	public static abstract class Receiver {

		/**
		 * 外部覆盖的方法，用于处理消息
		 * 
		 * @param msg
		 *            文本消息
		 * @return 如果为true，则继续接收消息，否则以后不再接收消息
		 */
		public abstract boolean excute(String msg,Map<String,String> properties);

	}

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		// TODO Auto-generated method stub

		log.info("dddd");
		String queueName = "Dmusp.log";
		String sendTextMsg = "sun超近11111";
		JmsPoolFactory.init();
		Map<String,String> properties=new HashMap<String,String>();
		properties.put("class", "34456");
		try {
			JmsUtils.send("default", queueName, sendTextMsg,properties);
		} catch (Exception e) {
			log.error("", e);
		}

		try {
			JmsUtils.send("default", queueName, sendTextMsg,properties);
		} catch (Exception e) {
			log.error("", e);
		}
		try {
			JmsUtils.send("default", queueName, sendTextMsg,properties);
		} catch (Exception e) {
			log.error("", e);
		}
		try {
			JmsUtils.send("default", queueName, sendTextMsg,properties);
		} catch (Exception e) {
			log.error("", e);
		}
		try {
			JmsUtils.send("default", queueName, sendTextMsg,properties);
		} catch (Exception e) {
			log.error("", e);
		}
		
		try {
			JmsUtils.send("default", queueName, sendTextMsg,properties);
		} catch (Exception e) {
			log.error("", e);
		}

		try {
			JmsUtils.send("default", queueName, sendTextMsg,properties);
		} catch (Exception e) {
			log.error("", e);
		}
		try {
			JmsUtils.send("default", queueName, sendTextMsg,properties);
		} catch (Exception e) {
			log.error("", e);
		}
		try {
			JmsUtils.send("default", queueName, sendTextMsg,properties);
		} catch (Exception e) {
			log.error("", e);
		}
		try {
			JmsUtils.send("default", queueName, sendTextMsg,properties);
		} catch (Exception e) {
			log.error("", e);
		}
		Thread.sleep(1000*2);
		JmsUtils.receiveAsyn("default", queueName, new Receiver() {

			@Override
			public boolean excute(String msg,Map<String,String> properties) {
				// TODO Auto-generated method stub
				System.out.println("msg="+msg+";"+ObjectUtils.toJsonString(properties));
				return true;
			}
		});

		JmsPoolFactory.clearAll(); 

		System.out.println("ok");

	}

}
